package com.study.kt._03coroutines

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import java.util.*

/**
 * <pre>
 *     author : slx
 *     time   : 2021/04/04
 *     desc   : 流的构建器:flow、flowOf可发送固定数量值的流、集合和序列都是asFlow扩展方法转换为Flow
 *
 *     Flow、Sequence中间操作符(filter、map等)，Flow可以在中间操作符的代码块中调用挂起函数
 * </pre>
 */
fun main() {
    runBlocking {
        // Flow中间操作
        (1..10).asFlow().map { input -> mapItem(input) }.collect { println(it) }

        flowOf("a", "b").transform { input ->
            // 流转换操作，针对每个元素执行下面操作
            emit(input.toUpperCase(Locale.ROOT))
            emit("finish")
        }.collect { println(it) }

        // take限定流数量：原理是达到指定数量后发出异常，中止流的发出
        flow {
            emit(1)
            emit(2)
            emit(3)
            emit(4)
        }.take(2).collect { println(it) }

        // 终止操作：如collect、toList、reduce等，终止操作都是挂起函数，终止操作才会真正开始执行流的收集
        // reduce 将多个结果汇集为一个结果
        val result = (1..4).asFlow().map { it * it }.reduce { a, b -> a + b }
        println(result)

        // 流的组合,组合完成后形成新的流
        val numFlow = (1..3).asFlow()
        val strFlow = flowOf("A", "B", "C")
        numFlow.zip(strFlow) { x, y -> "$x -> $y" }.collect { println(it) }

        // 流的展平,将该Flow的每个元素转换为新的Flow输出
        (1..3).asFlow().onEmpty { delay(100) }
                .flatMapConcat { flatFlow(it) }
                .collect { value ->
                    println(value)
                }
    }
}

private fun flatFlow(index: Int): Flow<String> = flow {
    emit("index start:$index")
    delay(200)
    emit("index end:$index")
}

private suspend fun mapItem(item: Int): String {
    delay(100)
    return "item:$item"
}